/**
 * Wihidum - Efficient Broker Hierarchy for Publish Subscribe Middleware
 *
 * License      :   Apache License 2.0
 * Home Page    :   http://code.google.com/p/wihidum/
 * Mailing List :   wihidum@googlegroups.com
 */

package lk.mrt.cse.wihidum.delivery;

import lk.mrt.cse.wihidum.storage.*;
import lk.mrt.cse.wihidum.processor.PublicationMessage;
import lk.mrt.cse.wihidum.util.SelfReader;
import org.apache.log4j.*;

import org.apache.axiom.soap.*;

import java.util.concurrent.*;
import java.util.HashMap;

/**
 * Pool of threads dedicated to send publications
 */
public class ThreadPool
        extends Thread
{

    private static final Logger logger = Logger.getLogger(ThreadPool.class);
    private static final int NO_OF_CONCURRENT_THREADS = 100;
    private static final ConcurrentLinkedQueue<PublicationMessage> publicationQueue = new ConcurrentLinkedQueue<PublicationMessage>();
    private ExecutorService executorService = Executors.newFixedThreadPool(NO_OF_CONCURRENT_THREADS);

    /**
     * Default constructor
     */
    public ThreadPool()
    {
        logger.debug("Thread pool created");
        this.start();
    }

    /**
     * Continuously check publication queue and send publications
     */
    public void run()
    {
        HashMap<String, String> brokerMap = SelfReader.read();
        while (true)
        {
            if (!publicationQueue.isEmpty())
            {
                PublicationMessage publicationMessage = this.dequeuePublication();

                executorService.execute(new PublicationThread(publicationMessage, brokerMap));
            }
        }
    }

    /**
     * Add a publication to the queue
     *
     * @param publicationMessage Publication to be added
     * @return True if added successfully, otherwise false
     */
    public boolean enqueuePublication(PublicationMessage publicationMessage)
    {
        logger.debug("Publication enqueued");
        return publicationQueue.add(publicationMessage);
    }

    /**
     * Remove a publication from the queue
     *
     * @return True if removed successfully, Otherwise false
     */
    private PublicationMessage dequeuePublication()
    {
        logger.debug("Publication dequeued");
        return publicationQueue.poll();
    }
}
